package com.example.toy.core.fs;

import com.example.toy.core.table.ToyTable;
import com.example.toy.core.utils.ToyPathDefine;
import com.example.toy.core.version.ToyVersion;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
import lombok.Data;
import lombok.SneakyThrows;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;

public class FsOfflineToyTableVersionFileView implements ToyTableVersionFileView {

  private final FileSystem fs;
  private final ToyVersion toyVersion;
  private final Path basePath;
  private ToyTable table;

  @SneakyThrows
  public FsOfflineToyTableVersionFileView(ToyTable table) {
    this.table = table;
    this.basePath = table.getTableMeta().getBasePath();
    this.toyVersion = table.getTableMeta().getVersion();
    this.fs = basePath.getFileSystem(new Configuration());
  }

  @Override
  public ToyVersion getVersion() {
    return this.toyVersion;
  }

  @SneakyThrows
  @Override
  public List<Path> getBaseFiles(String partitionKey) {
    Path partitionPath = table.getTableMeta().getPartitionPathMap().get(partitionKey);
    FileFilterLessThanVersion filter =
        new FileFilterLessThanVersion(table.getTableMeta().getVersion(), true);
    FileStatus[] fileStatuses = fs.listStatus(partitionPath, filter);

    List<ToyVersion> availableBaseVersions =
        Arrays.stream(fileStatuses)
            .map(FileStatus::getPath)
            .map(ToyPathDefine::extractVersion)
            .distinct()
            .sorted()
            .collect(Collectors.toList());
    if (availableBaseVersions.isEmpty()) {
      return new ArrayList<>();
    } else {
      ToyVersion latestBaseFileVersion =
          availableBaseVersions.get(availableBaseVersions.size() - 1);
      List<Path> baseFiles =
          Arrays.stream(fileStatuses)
              .map(FileStatus::getPath)
              .filter(
                  path -> {
                    ToyVersion toyVersion = ToyPathDefine.extractVersion(path);
                    return toyVersion.equals(latestBaseFileVersion);
                  })
              .collect(Collectors.toList());
      return baseFiles;
    }
  }

  @SneakyThrows
  @Override
  public List<Path> getDeltaFiles(String partitionKey) {
    Path partitionPath = table.getTableMeta().getPartitionPathMap().get(partitionKey);
    FileFilterLessThanVersion filter =
        new FileFilterLessThanVersion(table.getTableMeta().getVersion(), true);
    FileStatus[] fileStatuses = fs.listStatus(partitionPath, filter);

    List<ToyVersion> availableBaseVersions =
        Arrays.stream(fileStatuses)
            .map(FileStatus::getPath)
            .map(ToyPathDefine::extractVersion)
            .distinct()
            .sorted()
            .collect(Collectors.toList());

    ToyVersion latestBaseFileVersion =
        availableBaseVersions.isEmpty()
            ? ToyVersion.FIRST_VERSION
            : availableBaseVersions.get(availableBaseVersions.size() - 1);

    FileFilterLessThanVersion deltaFileFilter =
        new FileFilterLessThanVersion(table.getTableMeta().getVersion(), false);
    FileStatus[] deltaFiles = fs.listStatus(partitionPath, deltaFileFilter);
    List<Path> collect =
        Arrays.stream(deltaFiles)
            .map(FileStatus::getPath)
            .filter(
                path -> {
                  ToyVersion toyVersion = ToyPathDefine.extractVersion(path);
                  return toyVersion.bigThan(latestBaseFileVersion);
                })
            .collect(Collectors.toList());
    return collect;
  }

  @Override
  public boolean isArchived() {
    return false;
  }

  @Data
  static class FileFilterLessThanVersion implements PathFilter {
    private final ToyVersion version;
    private final boolean isBaseFile;

    @Override
    public boolean accept(Path path) {
      if (isBaseFile) {
        return ToyPathDefine.isBaseDataFile(path)
            && ToyPathDefine.extractVersion(path).smallThanOrEquals(version);
      } else {
        return ToyPathDefine.isDeltaDataFile(path)
            && ToyPathDefine.extractVersion(path).smallThanOrEquals(version);
      }
    }
  }
}
